package com.quectel.util.kit;

import com.quectel.util.ex.CheckErrorException;
import com.quectel.util.redis.RedisUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.function.BiConsumer;

/**
 * 同步数据库操作
 *
 * @author: caolu
 * @email: louis.cao@quectel.com
 * @date: 2023-03-01 15:30:00
 */
@Slf4j
public class SyncDataKit {

    private static final String LOCK_SUFFIX = "SYNC_DATA_KIT_LOCK";
    private static final String CACHE_SUFFIX = "SYNC_DATA_KIT_CACHE_MILLIS";


    /**
     * <pre>
     * 同步任务抽象时间会有重叠,注意业务数据去重
     * 例如: pullData("ALARM_DAYA_SYNC",3600,600,60,(startTime,endTime)->{})
     * 代表：创建一个名字为ALARM_DAYA_SYNC的任务 最多同步最近1小时并且最多允许迟到10分钟（此时我们最多取70分钟的数据），每1分钟执行一次
     * 场景：
     * 第一次定时任务在10:00开始执行 此时的回调时间为 08:50 10:00 缓存值为09:50
     *      定时任务在10:01开始执行 此时的回调时间为 09:50 10:01 缓存值为09:51
     *      定时任务在10:02开始执行 此时的回调时间为 09:51 10:02 缓存值为09:52
     *      也就是说时间会有重叠 重叠为10分钟 所以业务数据需要去重（Redis.setNX）处理
     * </pre>
     *
     * @param jobName               任务名称
     * @param maxSyncPeriodSeconds  最多取多少秒的数据
     * @param allowLateSeconds      允许数据迟到多少秒
     * @param taskExecPeriodSeconds 定时任务的执行周期 每30秒执行 传入30即可
     * @param consumer              开始时间和结束时间的回调
     */
    public static void pullData(
            String jobName,
            long maxSyncPeriodSeconds,
            long allowLateSeconds,
            long taskExecPeriodSeconds,
            BiConsumer<Date, Date> consumer
    ) {

        if (jobName == null || maxSyncPeriodSeconds <= 0 || allowLateSeconds <= 0 || taskExecPeriodSeconds <= 0 || consumer == null) {
            throw new CheckErrorException();
        }
        Lock spinLock = RedisUtils.getSpinLock(String.format("%s_%s", jobName, LOCK_SUFFIX));

        spinLock.lock();

        try {
            long currentMills = System.currentTimeMillis();
            long latestCacheMillis = currentMills - maxSyncPeriodSeconds - allowLateSeconds;

            String cacheKey = String.format("%s_%s", jobName, CACHE_SUFFIX);
            String cacheMills = RedisUtils.get(cacheKey);
            if (cacheMills != null) {
                long cacheValue = Long.parseLong(cacheMills);
                if ((currentMills - cacheValue) <= TimeUnit.SECONDS.toMillis(maxSyncPeriodSeconds + allowLateSeconds)) {
                    latestCacheMillis = cacheValue;
                }
            }

            consumer.accept(new Date(latestCacheMillis), new Date(currentMills));

            RedisUtils.set(cacheKey,
                    String.valueOf(currentMills - TimeUnit.SECONDS.toMillis(allowLateSeconds)),
                    taskExecPeriodSeconds + maxSyncPeriodSeconds + allowLateSeconds,
                    TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("Sync Data exception", e);
            throw e;
        } finally {
            spinLock.unlock();
        }


    }

}
